package com.index;

import java.io.IOException;
import java.io.StringReader;
import java.util.Iterator;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.wltea.analyzer.core.IKSegmenter;
import org.wltea.analyzer.core.Lexeme;

public class CountSplitWord {
	
	private static String path = "";
	


	public static class Map extends MapReduceBase implements Mapper<LongWritable,Text,Text,IntWritable> {
		private final static IntWritable one = new IntWritable(1);
	    private Text word = new Text();
		public void map(LongWritable key, Text value,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			StringReader reader=new StringReader(line);  
	        //分词  
			IKSegmenter seg = new IKSegmenter(reader,true);
			Lexeme lex=new Lexeme(0, 0, 0, 0);
	        while((lex=seg.next())!=null)  
	        {  
	        	word.set(lex.getLexemeText()+"#"+path);
	            output.collect(word,one);
	        }  
	        reader.close();
		}
	}
	
	public static class Combin extends MapReduceBase implements Reducer<Text,IntWritable,Text,IntWritable> {
		public void reduce(Text key, Iterator<IntWritable> values,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
				throws IOException {
			int sum = 0;
			while (values.hasNext()) {
				sum += values.next().get();
			}
			output.collect(key, new IntWritable(sum));
		}
	}
	/**
	 * 输出格式:word|allCount@path1|count&path2|count.......
	 * @author Administrator
	 *
	 */
	public static class Reduce extends MapReduceBase implements Reducer<Text,IntWritable,Text,NullWritable> {
		public void reduce(Text key, Iterator<IntWritable> values,
				OutputCollector<Text, NullWritable> output, Reporter reporter)
				throws IOException {
			int sum = 0;
			while (values.hasNext()) {
				sum += values.next().get();
			}
			String[] strs = key.toString().split("#");
			String msg = strs[0]+"#"+sum+"@"+strs[1]+"#"+sum;
			System.out.println("result : "+msg);
			output.collect(new Text(msg), null);
		}
	}
	
	
	public static String execute(String args0) throws IOException{
		System.out.println("begin cutting word count");
		path = args0;
		JobConf conf = new JobConf();
		conf.setJobName("CountSplitWord");
		
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(IntWritable.class);
		
		conf.setMapperClass(Map.class);
		conf.setCombinerClass(Combin.class);
		conf.setReducerClass(Reduce.class);
		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		FileInputFormat.addInputPath(conf, new Path(args0));
		String tmpFileName = String.valueOf(System.currentTimeMillis());
		Path out = new Path("/index/tmp/test.txt");
		FileSystem system = out.getFileSystem(conf);
		if(system.exists(out)){
			return tmpFileName;
		}
		FileOutputFormat.setOutputPath(conf, out);
		try{
			RunningJob job =JobClient.runJob(conf);
			job.isComplete();
			return tmpFileName;
		}catch(Exception e){
			return null;
		}
	}
}
